package com.o2o.cleaning.month.platform.ebusiness_plat.kuaishou.shuang11

import com.alibaba.fastjson.JSON
import com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.brand_join_res
import org.apache.spark.sql.{DataFrame, SparkSession, functions}

object kuaishou_02_OBS_TO_OBS {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .config("spark.debug.maxToStringFields", "500")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.caseSensitive", "true")
      .config("spark.executor.heartbeatInterval", "120s")
      .config("spark.network.timeout", "12000s")
      //      .config("spark.default.parallelism", "2")
      .config("es.nodes", "192.168.1.29")
      .config("es.index.read.missing.as.empty", "true")
      .config("es.net.http.auth.user", "elastic")
      .config("es.net.http.auth.pass", "changeme")
      .config("es.port", "9200")
      .config("cluster.name", "O2OElastic")
      .config("spark.sql.crossJoin.enabled", "true")
      //      .master("local[*]")
      //      .enableHiveSupport()
      .getOrCreate()

    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    //    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("WARN")

    //    val year = 2021
    val year = 2022
    //    val month = 11
    val month = 1111
    //    val timestamp = 1636560000
    val timestamp = 1668096000
    val platformName = "kuaishou"
    //        val address_source_path = s"s3a://o2o-dimension-table/address_table/address_table_${year}/${month}/address_platform/kuaishou_address_${year}_${month}/"
    val address_source_path = s"s3a://o2o-dimension-table/address_table/address_table_2022/11/address_platform/kuaishou_address_2022_11/"
    val datawork_path = s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/dws/${year}/1111/*"
    //    val datawork_path = "D://test.json"
    val classify_path = s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/categoryData/2021_9/"
    val result_path = s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/es/${year}/${month}/"
    //    val result_path = s"s3a://dws-data/g_data/kuaishou/dws/${year}/${month}/"

    println(s"${year}|${month}|${timestamp}")

    println("datawork_path -----> " + datawork_path)

    println("-----关联分类-----")
    val data_mid: DataFrame = Data_Get_Classify_Dw(spark, datawork_path, classify_path, timestamp)
    //    val data_mid: DataFrame = Data_Get_Classify_Dw(spark, "D:\\test.json", classify_path)
    data_mid.registerTempTable("t1")
    spark.sql(
      """
        |select count(1),sum(sellCount),sum(salesAmount) from t1
        |""".stripMargin).show(false)
    println("-----关联地址-----")
    val data_mid1 = kuaiShouAddress(spark, data_mid, address_source_path)
    data_mid1.registerTempTable("t2")
    spark.sql(
      """
        |select count(1),sum(sellCount),sum(salesAmount) from t2
        |""".stripMargin).show(false)
    data_mid1.printSchema()
    data_mid1.coalesce(30).write.mode("overwrite").orc(result_path)

    //    println("-----关联品牌-----")
    //    val brand = new brand_join_res
    //    brand.brandJoinResult(data_mid1, result_path, year, month, platformName, spark)
    //    sc.stop()
    //    spark.stop()

  }


  def Data_Get_Classify_Dw(sc: SparkSession, datawork_path: String, classify_path: String, timestamp: Int): DataFrame = {

    sc.read.orc(datawork_path)
      .drop("anchor_count")
      .drop("coupon_list")
      .drop("creditlevel")
      .drop("favcount")
      .drop("fifthcategoryid")
      .drop("final_price")
      .drop("final_price_int")
      .drop("flagship_type")
      .drop("is_jhs")
      .drop("is_live_good")
      .drop("is_oldfornew")
      .drop("is_onsell")
      .drop("lastCategoryId")
      .drop("lastCategoryName")
      .drop("live_count")
      .drop("price_abnormal_degree")
      .drop("price_int")
      .drop("quantity")
      .drop("random_int_val")
      .drop("shop_live_count")
      .drop("shopimages")
      .drop("standardlastcategoryid")
      .drop("visitor_count")
      .drop("starts")
      .registerTempTable("t_sell") //数据
    sc.sql(
      """
        |select good_id
        |,max(platform_id) platform_id
        |,'' as title
        |,max(images) images
        |,sum(sellcount) sellcount
        |,sum(salesamount) salesamount
        |,max(price) price
        |,max(original_cost) original_cost
        |,max(fans) fans
        |,max(allitemcount) allitemcount
        |,max(evaluates) evaluates
        |,max(goodratepercentage) goodratepercentage
        |,max(commentcount) commentcount
        |,max(is_newproduct) is_newproduct
        |,max(goodurl) goodurl
        |,'' as base_info
        |,max(originproduction) originproduction
        |,max(originitem) originitem
        |,max(origincountryname) origincountryname
        |,max(taxfreezone) taxfreezone
        |,max(bc_type) bc_type
        |,max(is_oversea) is_oversea
        |,max(is_market) is_market
        |,max(shopid) shopid
        |,max(userid) userid
        |,max(shopname) shopname
        |,max(shopurl) shopurl
        |,max(is_flagship) is_flagship
        |,max(is_selfsupport) is_selfsupport
        |,max(company_name) company_name
        |,max(region_id) region_id
        |,max(province) province
        |,max(city) city
        |,max(district) district
        |,max(brand_id) brand_id
        |,max(brand_name) brand_name
        |,max(standard_brand_id) standard_brand_id
        |,max(standard_brand_name) standard_brand_name
        |,max(brand_state) brand_state
        |,max(brand_type) brand_type
        |,max(is_laozihao) is_laozihao
        |,max(laozihao_type) laozihao_type
        |,max(parentbrandid) parentbrandid
        |,max(lastcategoryid) lastcategoryid
        |,max(lastcategoryname) lastcategoryname
        |,max(rootcategoryid) rootcategoryid
        |,max(rootcategoryname) rootcategoryname
        |,max(categoryid) categoryid
        |,max(categoryname) categoryname
        |,max(subcategoryid) subcategoryid
        |,max(subcategoryname) subcategoryname
        |,max(detailcategoryid) detailcategoryid
        |,max(detailcategoryname) detailcategoryname
        |,max(firstcategoryid) firstcategoryid
        |,max(secondcategoryid) secondcategoryid
        |,max(thirdcategoryid) thirdcategoryid
        |,max(fourthcategoryid) fourthcategoryid
        |,max(timestamp) timestamp
        |,max(brand_company_name) brand_company_name
        |,max(address) address
        |,max(mo) mo
        |from t_sell group by good_id
        |""".stripMargin).registerTempTable("shop_data_table")

    sc.sql("select sum(sellcount),sum(salesamount) from shop_data_table").show(false)
    sc.read.option("header", "true").csv(classify_path).createOrReplaceTempView("classify_table") //分类映射表

    val data_mid1 = sc.sql(
      """
        |SELECT shop.*,
        |IFNULL(classify.firstCategoryId, "10099") firstCategoryId_c,
        |IFNULL(classify.secondCategoryId , "1009999") secondCategoryId_c,
        |IFNULL(classify.thirdCategoryId, "100999999") thirdCategoryId_c,
        |IFNULL(classify.fourthCategoryId,"10099999999") fourthCategoryId_c
        |FROM
        |(SELECT * ,
        |    IFNULL(lastcategoryid, "10099") AS lastCategoryId_c
        |    ,IFNULL(lastcategoryname, "-1") AS lastCategoryName_c
        |FROM
        | shop_data_table
        |) shop
        |LEFT JOIN
        |(
        |SELECT lastcategoryid , firstCategoryId , secondCategoryId , thirdCategoryId , fourthCategoryId
        |FROM classify_table
        |) classify ON shop.lastCategoryId_c = classify.lastcategoryid
        |where classify.lastcategoryid is not null
        |""".stripMargin).drop("lastcategoryid", "lastcategoryname", "firstcategoryid", "secondcategoryid", "thirdcategoryid", "fourthcategoryid")

    sc.sql(
      """
        |SELECT shop.*
        |FROM
        |(SELECT * ,
        |    IFNULL(lastcategoryid, "10099") AS lastCategoryId_c
        |    ,IFNULL(lastcategoryname, "-1") AS lastCategoryName_c
        |FROM
        | shop_data_table
        |) shop
        |LEFT JOIN
        |(
        |SELECT lastcategoryid , firstCategoryId , secondCategoryId , thirdCategoryId , fourthCategoryId
        |FROM classify_table
        |) classify ON shop.lastCategoryId_c = classify.lastcategoryid
        |where classify.lastcategoryid is null
        |""".stripMargin)
      .registerTempTable("mid1")
    val data_mid2 = sc.sql(
      """
        |SELECT shop.*,
        |IFNULL(classify.firstCategoryId, "10099") firstCategoryId_c,
        |IFNULL(classify.secondCategoryId , "1009999") secondCategoryId_c,
        |IFNULL(classify.thirdCategoryId, "100999999") thirdCategoryId_c,
        |IFNULL(classify.fourthCategoryId,"10099999999") fourthCategoryId_c
        |FROM
        |(SELECT *
        |FROM
        | mid1
        |) shop
        |LEFT JOIN
        |(
        |SELECT lastcategoryid , firstCategoryId , secondCategoryId , thirdCategoryId , fourthCategoryId
        |FROM classify_table
        |) classify ON shop.lastCategoryId_c = classify.thirdCategoryId
        |where classify.thirdCategoryId is not null
        |""".stripMargin).drop("lastcategoryid", "lastcategoryname", "firstcategoryid", "secondcategoryid", "thirdcategoryid", "fourthcategoryid")

    sc.sql(
      """
        |SELECT shop.*
        |FROM
        |(SELECT *
        |FROM
        | mid1
        |) shop
        |LEFT JOIN
        |(
        |SELECT lastcategoryid , firstCategoryId , secondCategoryId , thirdCategoryId , fourthCategoryId
        |FROM classify_table
        |) classify ON shop.lastCategoryId_c = classify.thirdCategoryId
        |where classify.thirdCategoryId is null
        |""".stripMargin)
      .registerTempTable("mid2")


    val data_mid3 = sc.sql(
      """
        |SELECT shop.*,
        |IFNULL(classify.firstCategoryId, "10099") firstCategoryId_c,
        |IFNULL(classify.secondCategoryId , "1009999") secondCategoryId_c,
        |IFNULL(classify.thirdCategoryId, "100999999") thirdCategoryId_c,
        |IFNULL(classify.fourthCategoryId,"10099999999") fourthCategoryId_c
        |FROM
        |(SELECT *
        |FROM
        | mid2
        |) shop
        |LEFT JOIN
        |(
        |SELECT lastcategoryid , firstCategoryId , secondCategoryId , thirdCategoryId , fourthCategoryId
        |FROM classify_table
        |) classify ON shop.lastCategoryId_c = classify.secondCategoryId
        |where classify.secondCategoryId is not null
        |""".stripMargin).drop("lastcategoryid", "lastcategoryname", "firstcategoryid", "secondcategoryid", "thirdcategoryid", "fourthcategoryid")

    sc.sql(
      """
        |SELECT shop.*
        |FROM
        |(SELECT *
        |FROM
        | mid2
        |) shop
        |LEFT JOIN
        |(
        |SELECT lastcategoryid , firstCategoryId , secondCategoryId , thirdCategoryId , fourthCategoryId
        |FROM classify_table
        |) classify ON shop.lastCategoryId_c = classify.secondCategoryId
        |where classify.secondCategoryId is null
        |""".stripMargin)
      .registerTempTable("mid3")

    val data_mid4 = sc.sql(
      """
        |SELECT shop.*,
        |IFNULL(classify.firstCategoryId, "10099") firstCategoryId_c,
        |IFNULL(classify.secondCategoryId , "1009999") secondCategoryId_c,
        |IFNULL(classify.thirdCategoryId, "100999999") thirdCategoryId_c,
        |IFNULL(classify.fourthCategoryId,"10099999999") fourthCategoryId_c
        |FROM
        |(SELECT *
        |FROM
        | mid3
        |) shop
        |LEFT JOIN
        |(
        |SELECT lastcategoryid , firstCategoryId , secondCategoryId , thirdCategoryId , fourthCategoryId
        |FROM classify_table
        |) classify ON shop.lastCategoryId_c = classify.firstCategoryId
        |""".stripMargin).drop("lastcategoryid", "lastcategoryname", "firstcategoryid", "secondcategoryid", "thirdcategoryid", "fourthcategoryid")

    data_mid1.unionAll(data_mid2).unionAll(data_mid3).unionAll(data_mid4).registerTempTable("data_classify_table")

    val data_classify = sc.sql(
      """
        |SELECT * FROM
        |data_classify_table
        |WHERE sellcount  > 0
        |""".stripMargin).toJSON.rdd
      .map(x => {
        val json = JSON.parseObject(x.toString)
        val brand_name = json.getOrDefault("brand_name", "-1").toString
        val firstcategoryid = json.getOrDefault("firstCategoryId_c", "10099").toString
        val secondcategoryid = json.getOrDefault("secondCategoryId_c", "1009999").toString
        val thirdcategoryid = json.getOrDefault("thirdCategoryId_c", "100999999").toString
        val fourthcategoryid = json.getOrDefault("fourthCategoryId_c", "10099999999").toString
        val fans = json.getOrDefault("fans", "").toString
        val allitemcount = json.getOrDefault("allitemcount", "").toString
        val goodratepercentage = json.getOrDefault("goodratepercentage", "").toString
        val commentcount = json.getOrDefault("commentcount", "").toString
        val originproduction = json.getOrDefault("originproduction", "").toString
        val originitem = json.getOrDefault("originitem", "").toString
        val origincountryname = json.getOrDefault("origincountryname", "").toString
        val taxfreezone = json.getOrDefault("taxfreezone", "").toString
        val bc_type = json.getOrDefault("bc_type", "").toString
        val userid = json.getOrDefault("userid", "").toString
        val standard_brand_id = json.getOrDefault("standard_brand_id", "").toString
        val standard_brand_name = json.getOrDefault("standard_brand_name", "").toString
        val is_laozihao = json.getOrDefault("is_laozihao", "").toString
        val laozihao_type = json.getOrDefault("laozihao_type", "").toString

        json.put("brand_name", brand_name)
        json.put("firstCategoryId_c", firstcategoryid)
        json.put("secondCategoryId_c", secondcategoryid)
        json.put("thirdCategoryId_c", thirdcategoryid)
        json.put("fourthCategoryId_c", fourthcategoryid)
        json.put("fans", fans)
        json.put("allitemcount", allitemcount)
        json.put("goodratepercentage", goodratepercentage)
        json.put("commentcount", commentcount)
        json.put("originproduction", originproduction)
        json.put("originitem", originitem)
        json.put("origincountryname", origincountryname)
        json.put("taxfreezone", taxfreezone)
        json.put("bc_type", bc_type)
        json.put("userid", userid)
        json.put("standard_brand_id", standard_brand_id)
        json.put("standard_brand_name", standard_brand_name)
        json.put("is_laozihao", is_laozihao)
        json.put("laozihao_type", laozihao_type)

        json.toString
      })


    val rdd = sc.read.json(data_classify)
      .withColumn("platformName", functions.lit("快手"))
      .withColumn("timeStamp", functions.lit(s"${timestamp}"))
      .withColumnRenamed("brand_id", "brandValueId")
      .withColumnRenamed("brand_name", "brandName")
      .withColumnRenamed("firstCategoryId_c", "firstCategoryId")
      .withColumnRenamed("secondCategoryId_c", "secondCategoryId")
      .withColumnRenamed("thirdCategoryId_c", "thirdCategoryId")
      .withColumnRenamed("fourthCategoryId_c", "fourthCategoryId")
      .withColumnRenamed("bc_type", "shopType")
      .withColumnRenamed("platform_id", "platformId")
      .withColumnRenamed("base_info", "Base_Info")
      .withColumnRenamed("categoryid", "categoryId")
      .withColumnRenamed("categoryname", "categoryName")
      .withColumnRenamed("commentcount", "commentCount")
      .withColumnRenamed("goodratepercentage", "goodRatePercentage")
      .withColumnRenamed("goodurl", "goodUrl")
      .withColumnRenamed("is_newproduct", "is_newProduct")
      .withColumnRenamed("is_selfsupport", "is_selfSupport")
      .withColumnRenamed("origincountryname", "originCountryName")
      .withColumnRenamed("originitem", "originItem")
      .withColumnRenamed("originproduction", "originProduction")
      .withColumnRenamed("rootcategoryid", "rootCategoryId")
      .withColumnRenamed("rootcategoryname", "rootCategoryName")
      .withColumnRenamed("salesamount", "salesAmount")
      .withColumnRenamed("sellcount", "sellCount")
      .withColumnRenamed("shopid", "shopId")
      .withColumnRenamed("shopurl", "shopUrl")
      .withColumnRenamed("subcategoryid", "subCategoryId")
      .withColumnRenamed("subcategoryname", "subCategoryName")
      .withColumnRenamed("taxfreezone", "taxFreeZone")
      .withColumnRenamed("userid", "userId")
      .withColumnRenamed("shopname", "shopName")
      .withColumnRenamed("allitemcount", "allItemCount")
      .withColumnRenamed("brand_state", "brand_state")
      .withColumnRenamed("brand_type", "brand_type")
      .withColumnRenamed("is_laozihao", "brand_isLaoZiHao")
      .withColumnRenamed("standard_brand_name", "brandName_cn")
      .drop("timestamp")
      .drop("lastCategoryId_c")
      .drop("lastCategoryName_c")
      .drop("region_id", "province", "city", "district")
      .drop("laozihao_type", "standard_brand_id")
      .drop("mo") //.registerTempTable("data_cate_table")

    rdd

  }

  // 处理地址
  def kuaiShouAddress(spark: SparkSession, frame: DataFrame, address_source_path: String): DataFrame = {
    frame.drop("address")
      .drop("administrative_region")
      .drop("aedzId")
      .drop("city")
      .drop("city_grade")
      .drop("city_origin")
      .drop("district")
      .drop("district_origin")
      .drop("economic_division")
      .drop("if_city")
      .drop("if_district")
      .drop("if_state_level_new_areas")
      .drop("latitude")
      .drop("longitude")
      .drop("name")
      .drop("poor_counties")
      .drop("province")
      .drop("regional_ID")
      .drop("registration_institution")
      .drop("rural_demonstration_counties")
      .drop("rural_ecommerce")
      .drop("the_belt_and_road_city")
      .drop("the_belt_and_road_province")
      .drop("the_yangtze_river_economic_zone_city")
      .drop("the_yangtze_river_economic_zone_province")
      .drop("town")
      .drop("urban_agglomerations")
      .registerTempTable("all")
    spark.read.json(address_source_path).registerTempTable("address")
    spark.sql(
      """
        |select t1.*,
        |case when t2.shopId is not null then t2.address  else  '0' end address,
        |case when t2.shopId is not null then t2.administrative_region  else  '0' end administrative_region,
        |case when t2.shopId is not null then t2.aedzId  else  '0' end aedzId,
        |case when t2.shopId is not null then t2.city  else  '0' end city,
        |case when t2.shopId is not null then t2.city_grade  else  '0' end city_grade,
        |case when t2.shopId is not null then t2.city_origin  else  '0' end city_origin,
        |case when t2.shopId is not null then t2.district  else  '0' end district,
        |case when t2.shopId is not null then t2.district_origin  else  '0' end district_origin,
        |case when t2.shopId is not null then t2.economic_division  else  '0' end economic_division,
        |case when t2.shopId is not null then t2.if_city  else  '0' end if_city,
        |case when t2.shopId is not null then t2.if_district  else  '0' end if_district,
        |case when t2.shopId is not null then t2.if_state_level_new_areas  else  '0' end if_state_level_new_areas,
        |case when t2.shopId is not null then t2.latitude  else  '0' end latitude,
        |case when t2.shopId is not null then t2.longitude  else  '0' end longitude,
        |case when t2.shopId is not null then t2.name  else  '0' end name,
        |case when t2.shopId is not null then t2.poor_counties  else  '0' end poor_counties,
        |case when t2.shopId is not null then t2.province  else  '0' end province,
        |case when t2.shopId is not null then t2.regional_ID  else  '0' end regional_ID,
        |case when t2.shopId is not null then t2.registration_institution  else  '0' end registration_institution,
        |case when t2.shopId is not null then t2.rural_demonstration_counties  else  '0' end rural_demonstration_counties,
        |case when t2.shopId is not null then t2.rural_ecommerce  else  '0' end rural_ecommerce,
        |-- case when t2.shopId is not null then t2.shopId  else  '0' end shopId,
        |case when t2.shopId is not null then t2.the_belt_and_road_city  else  '0' end the_belt_and_road_city,
        |case when t2.shopId is not null then t2.the_belt_and_road_province  else  '0' end the_belt_and_road_province,
        |case when t2.shopId is not null then t2.the_yangtze_river_economic_zone_city  else  '0' end the_yangtze_river_economic_zone_city,
        |case when t2.shopId is not null then t2.the_yangtze_river_economic_zone_province  else  '0' end the_yangtze_river_economic_zone_province,
        |case when t2.shopId is not null then t2.town  else  '0' end town,
        |case when t2.shopId is not null then t2.urban_agglomerations  else  '0' end urban_agglomerations
        |from all t1 left join address t2
        |on t1.shopId = t2.shopId
        |""".stripMargin).drop("shopName").registerTempTable("t_all2")
    // 这一步是处理 相同的 shopId 对应不同的shopName
    spark.sql(
      """
        |select shopName,shopId from
        |(select shopName,shopId,rank() over(partition by shopId order by sum(salesAmount) desc) rank
        |from all
        |group by shopName,shopId)
        |where rank = 1
        |""".stripMargin)
      .registerTempTable("t_shopName")
    val frameAddress = spark.sql(
      """
        |select t1.*,t2.shopName
        |from t_all2 t1
        |left join t_shopName t2
        |on t1.shopId = t2.shopId
        |""".stripMargin)
    frameAddress
  }
}
